#ifndef EX04_NESTED
#include <deque>
#include <algorithm>
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <assert.h>
#include "pwext.hpp"
#endif // EX04_NESTED

struct Message 
{
	Message(int mtype = NORMAL) : msgtype_(mtype)
	{ }
	virtual ~Message() { } // allow for polymorphism

	enum {
		EXIT,
		NORMAL, // first normal type
#ifndef EX04_NESTED
		SOME_MSG_1
#endif // EX04_NESTED
	};
	int msgtype_;
	// ... whatever else is needed by the application ...
};

// [[ex04dc]]
// DEBUG descr The queue with two message priority levels.
class Queue
{
protected:
	enum post_l_result {
		P_FAIL,
		P_SUCCESS,
		P_POSTIT
	};

public:
	enum Priority {
		PRIORITY_HIGH = 0,
		// all the other values are treated as NORMAL
		PRIORITY_NORMAL 
	};

	enum {
		DEFAULT_LIMIT = 1000
	};

	// use the AllExitQueue approach, with only the
	// "forwards" connections counted for the posting
	// of EXIT
	Queue(int limit = DEFAULT_LIMIT) :
		limit_(limit), 
		condrd_(condwr_), priority_not_empty_(false),
		exitposted_(false), nwriters_(0)
	{ 
		int i = 0;
		dqwr_[PRIORITY_HIGH] = &deques_[i++];
		dqrd_[PRIORITY_HIGH] = &deques_[i++];
		dqwr_[PRIORITY_NORMAL] = &deques_[i++];
		dqrd_[PRIORITY_NORMAL] = &deques_[i++];
	}

	virtual ~Queue()
	{ 
		Message *m;
		for (int i = 0; i < 4; i++) {
			while (!deques_[i].empty()) {
				m = deques_[i].front();
				deques_[i].pop_front();
				delete m;
			}
		}
	}

	// virtual in case if later a subclass wants to do
	// something different; returns true on success,
	// or false if the handler has already been told
	// to exit
	virtual bool add_writer()
	{ 
		pw::lockmutex lm(mutexx_);
		if (exitposted_)
			return false;
		++nwriters_;
		return true;
	}

	// this combines the STL front() and pop_front()
	// since the popping must be atomic; the receiver
	// takes over the ownership of the message
	Message *pop_front()
	{
		pw::lockmutex lm(mutexrd_);

		// if the reading of NORMAL dqrd was in progress
		// but a PRIORITY message arrived, must switch
		// to reading the PRIORITY messages.
		if (priority_not_empty_ 
		&& dqrd_[PRIORITY_HIGH]->empty()) {
			pw::lockmutex lm(condwr_);
			swap(dqwr_[PRIORITY_HIGH], dqrd_[PRIORITY_HIGH]);
			priority_not_empty_ = false;
		}

		while(dqrd_[PRIORITY_HIGH]->empty()
		&& dqrd_[PRIORITY_NORMAL]->empty()) {
			pw::swapmutex sw(mutexrd_, condwr_);

			if (dqwr_[PRIORITY_HIGH]->empty()
			&& dqwr_[PRIORITY_NORMAL]->empty()) {
				// it releases and gets back the mutex
				condrd_.wait(); 
			} else {
				if (!dqwr_[PRIORITY_HIGH]->empty()) {
					swap(dqwr_[PRIORITY_HIGH], 
						dqrd_[PRIORITY_HIGH]);
					// high-priority posting doesn't block, 
					// no need to wake up the writers
				}
				if (!dqwr_[PRIORITY_NORMAL]->empty()) {
					swap(dqwr_[PRIORITY_NORMAL], 
						dqrd_[PRIORITY_NORMAL]);
					// if the write buffer was full, wake up
					// the writers
					condwr_.broadcast();
				}
			}
		}

		Message *m;
		if(!dqrd_[PRIORITY_HIGH]->empty()) {
			m = dqrd_[PRIORITY_HIGH]->front();
			dqrd_[PRIORITY_HIGH]->pop_front();
		} else {
			m = dqrd_[PRIORITY_NORMAL]->front();
			dqrd_[PRIORITY_NORMAL]->pop_front();
		}
		return m;
	}

	// takes over the ownership of the message;
	// returns true if the message was posted, false
	// if discarded
	virtual bool post(Priority pri, Message *m)
	{
		if (pri != PRIORITY_HIGH)
			pri = PRIORITY_NORMAL;

		int res;
		{
			pw::lockmutex lm(mutexx_);
			res = post_l(pri, m);
		}
		if (res == P_POSTIT)
			return push_back(pri, m);
		else
			return (res != P_FAIL);
	}

protected:

	virtual post_l_result post_l(Priority pri, Message *m)
	{
		if (exitposted_) {
			delete m;
			return P_FAIL;
		}
		if (m->msgtype_ == Message::EXIT) {
			if (pri == PRIORITY_HIGH) {
				delete m;
				// can't send EXIT in a priority message
				return P_FAIL; 
			}

			if (--nwriters_ <= 0) {
				// last writer, post EXIT to the handler
				exitposted_ = true;
			} else {
				// message is processed successfully but it
				// has nothing to pass to the handler yet
				delete m;
				return P_SUCCESS;
			}
		}
		return P_POSTIT;
	}

	// returns false if the exit was already posted,
	// and this is not the EXIT message itself (then
	// the message also gets discarded);
	// expects that the priority value is in the
	// correct range
	bool push_back(Priority pri, Message *m)
	{
		bool need_prodding, result;
		{
			pw::lockmutex lm(condwr_);
			result = push_back_l(pri, m, need_prodding);
		}
		if (need_prodding) {
			pw::lockmutex lm(mutexrd_);
			priority_not_empty_ = true;
		}
		return result;
	}
	// Caller must have the mutex in condwr_ locked.
	// If need_prodding is returned true, the caller
	// must set the flag priority_not_empty_ with
	// locked mutexrd_.
	bool push_back_l(Priority pri, Message *m, 
		bool &need_prodding)
	{
		need_prodding = false;
		if (exitposted_ && m->msgtype_ != Message::EXIT) {
			delete m;
			return false;
		}

		if (pri == PRIORITY_HIGH) {
			// remember if the reader may need prodding to
			// switch from the buffer of normal messages to
			// priority ones
			need_prodding = dqwr_[PRIORITY_HIGH]->empty();
		} else {
			// flow control only for the normal messages
			while (dqwr_[PRIORITY_NORMAL]->size() >= limit_)
				// it releases and gets back the mutex
				condwr_.wait(); 
		}

		dqwr_[pri]->push_back(m);
		condrd_.broadcast();
		return true;
	}

	int limit_;
	std::deque<Message *> deques_[4];
	pw::pmutex mutexrd_;
	std::deque<Message *> *dqwr_[2], *dqrd_[2];
	pw::pmcond condwr_;
	pw::pchaincond condrd_;
	// flag: the priority dqwr became non-empty
	bool priority_not_empty_; 
	// for the exit accounting
	pw::pmutex mutexx_; 
	// whether the EXIT message was posted to the queue
	bool exitposted_; 
	// number of writers that haven't already exited
	int nwriters_; 
};
// END

#ifndef EX04_NESTED
// DEBUG {

struct data : public Message
{
	int v_;

	data(int v, int kind = Message::NORMAL) : Message(kind), v_(v)
	{ }
	~data()
	{
		fprintf(stderr, "deleted data %d\n", v_);
	}
	virtual int getData()
	{
		return v_;
	}
};

// paces the handler
pw::semaphore hsem(0);

class Handler : public pw::pwthread
{
};

class SomeHandler : public Handler
{
	friend class SomeQueue;
public:
	void setMyQueue(Queue *mine)
	{
		q_ = mine;
	}
	void setNextQueue(Queue *next)
	{
		nextq_ = next;
	}
	// ...
	void *execute()
	{
		while(true) {
			hsem.wait(1);
			Message *m = q_->pop_front();
			{
				pw::lockmutex lm(mutex_);

				switch (m->msgtype_) {
				case Message::EXIT:
					 fprintf(stderr, "Process exit\n");
					nextq_->post(Queue::PRIORITY_NORMAL, m);
					return 0;
				case Message::NORMAL:
					// ... process it ...
					 fprintf(stderr, "Process normal %d\n", ((data *)m)->getData());
					delete m;
					break;
				}
			}
		}
	}

protected:
	Queue *q_;
	Queue *nextq_; // input queue of the next handler in pipeline
	pw::pmutex mutex_;
};

int main()
{
	SomeHandler h;
	Queue q;
	q.add_writer();

	Queue qnext;
	qnext.add_writer();

	h.setMyQueue(&q);
	h.setNextQueue(&qnext);

	h.start();

	q.post(Queue::PRIORITY_NORMAL, new data(1));
	q.post(Queue::PRIORITY_HIGH, new data(2));
	q.post(Queue::PRIORITY_NORMAL, new Message(Message::EXIT));
	hsem.signal(10);
	h.join();

	return 0;
}
/* Sample output (timing of URGENT processing may vary):
Process URGENT_1 3
deleted data 3
Start process normal 1
End process normal 1
deleted data 1
Start process normal 2
End process normal 2
deleted data 2
Process exit
deleted data 99
*/
// DEBUG }
#endif // EX04_NESTED
